package com.atguigu0.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @description: sparkStreaming案例,无状态,每次统计单词的个数不会进行合并,5s内数据互不相干.无状态转换
 * @time: 2020/6/15 20:54
 * @author: baojinlong
 **/
object WordCountStreaming2 {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    // 创建SteamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 创建Dstream
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    // 转换rdd操作
    val wordToCountDStream: DStream[(String, Int)] = line.transform(rdd => {
      // 压平,转成rdd后就可以根据需要转化成我们需要的格式如 DataFrame,DataSet等
      val word: RDD[String] = rdd.flatMap(_.split(" "))
      val wordToOne: RDD[(String, Int)] = word.map((_, 1))
      val wordToCount: RDD[(String, Int)] = wordToOne.reduceByKey(_ + _)
      wordToCount
    })
    // 打印数据
    wordToCountDStream.print()
    // 开启sparkStreaming
    ssc.start()
    // 等待执行
    ssc.awaitTermination()
  }
}
